package com.luoqifei.yomob.realtime.analysis.kafka.stream;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * Created by luoqifei on 2017/09/27.
 */
public class YomobKafkaClient {
    public static final Logger LOG = LoggerFactory.getLogger(YomobKafkaClient.class);
    private KafkaProducer<String, String> producer;
    private KafkaConsumer<Object, Object> consumer;
    private Properties defaultProducerProps;
    private Properties defaultConsumerProps;

    /**
     * @param config
     * @deprecated
     */
    public YomobKafkaClient(Properties config) {
        this.producer = new KafkaProducer<String, String>(config);
        this.consumer = new KafkaConsumer<Object, Object>(config);
    }

    public YomobKafkaClient() {
    }

    public void initProducer(Properties config) {
        this.producer = new KafkaProducer<String, String>(config);
        LOG.debug("success to init producer by config {}",config.toString());
    }

    public void initProducer(String serverAddress) {
        defaultProducerProps = new Properties();
        defaultProducerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddress);
        initProducerDefaultProps();
        this.producer = new KafkaProducer<String, String>(defaultProducerProps);
        LOG.debug("success to init producer by default config {}",defaultProducerProps.toString());

        //testKafka();
    }

    public void initConsumer(Properties config) {
        KafkaConsumer<Object, Object> consumer = new KafkaConsumer<>(config);
        LOG.debug("success to init consumer by config {}",config.toString());
    }

    public void initConsumer(String serverAddress) {
        defaultConsumerProps = new Properties();
        defaultConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddress);
        initConsumerDefaultProps();
        this.consumer = new KafkaConsumer<Object, Object>(defaultConsumerProps);
        LOG.debug("success to init consumer by default config {}",defaultConsumerProps.toString());

    }

    public KafkaProducer<String, String> getProducer() {
        return producer;
    }

    public KafkaConsumer<Object, Object> getConsumer() {
        return consumer;
    }

    /**
     * @throws ExecutionException
     * @throws InterruptedException
     * @deprecated
     */
    private void testKafka() throws ExecutionException, InterruptedException {
        //producer.send(new ProducerRecord<String, String>("test", "health test", "health test")).get();
    }

    private void initProducerDefaultProps() {
        defaultProducerProps.put(ProducerConfig.ACKS_CONFIG, "1");
        defaultProducerProps.put(ProducerConfig.RETRIES_CONFIG, 3);
        defaultProducerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        defaultProducerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    }

    private void initConsumerDefaultProps() {
        defaultConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        defaultConsumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        defaultConsumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        defaultConsumerProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        defaultConsumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        defaultConsumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
        defaultConsumerProps.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 1000);
    }


    public void close() {
        if (this.producer != null) {
            producer.close();
            LOG.debug("success closed producer.");
        }
        if (this.consumer != null) {
            consumer.close();
            LOG.debug("success closed consumer.");
        }
        LOG.debug("success closed YomobKafkaClient.");
    }

    public static void main(String[] args) throws Exception {
        if (args.length == 0) {
            System.out.println("Enter topic name");
            return;
        }
        String topicName = args[0].toString();
        topicName = "dau-source";
        Properties props = new Properties();
        //Assign localhost id
        props.put("bootstrap.servers", "localhost:9092");
        //Set acknowledgements for producer requests.
        props.put("acks", "all");
        //If the request fails, the producer can automatically retry,
        props.put("retries", 0);
        //Specify buffer size in config
        props.put("batch.size", 16384);
        //Reduce the no of requests less than 0
        props.put("linger.ms", 1);

        //The buffer.memory controls the total amount of memory available to the producer for buffering.
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", StringSerializer.class);
        props.put("value.serializer", StringSerializer.class);
        String recordValue="{\"appid\":\"3u0AL6oi1vx0T6vn4oV9\",\"publisherid\":\"10053\",\"channelid\":\"10053\",\"userid\":\"849532892348420\",\"counterid\":\"active\",\"version\":\"1.6.8p2\",\"platform\":\"2\",\"time\":\"1506068229\",\"metadata\":\"{\\\"bundle_id\\\":\\\"com.tf.sse\\\",\\\"country_id\\\":\\\"42\\\",\\\"udid\\\":\\\"f07fcc4e74d00500\\\",\\\"net\\\":\\\"wifi\\\",\\\"system\\\":\\\"samsung_SM-G9008V_7.1.2\\\",\\\"bundleid\\\":\\\"com.tf.sse\\\"}\"}";

        Producer<String, String> producer = new KafkaProducer
                <String, String>(props);
        // producer.send(new ProducerRecord<String, String>(topicName,"value","55"));
        for (int i = 10; i < 11; i++)
            producer.send(new ProducerRecord<String, String>(topicName,
                    Integer.toString(i), recordValue));
        System.out.println("Message sent successfully");
        producer.close();
    }
}